
package wal

import (
	"encoding/binary"
	"hash"
	"io"
	"os"
	"sync"

	"go.etcd.io/etcd/pkg/v3/crc"
	"go.etcd.io/etcd/pkg/v3/ioutil"
	"go.etcd.io/etcd/server/v3/storage/wal/walpb"
)

// walPageBytes是将记录刷新到后台写入程序的对齐方式。
// 它应该是最小扇区大小的倍数，以便WAL可以安全地
// 区分损坏写入和普通数据损坏。
const walPageBytes = 8 * minSectorSize

// encoder.encode（）方法是真正完成日志写入功能的地方
type encoder struct {
	mu sync.Mutex	// 在进行读写文件的时候，需要加锁同步。
	// PageWriter是带有缓冲区的Writer，在写入时，每写满一个Page大小的缓冲区，
	// 就会自动触发一次Flush操作，将数据同步刷新到磁盘上。每个Page的大小是由walPageBytes常量指定的。
	bw *ioutil.PageWriter
	crc       hash.Hash32
	// 日志序列化之后，会暂存在该缓冲区中，该缓冲区会被复用，这就防止了每次序列化创建缓冲区带来的开销。
	buf       []byte
	// 在写入一条日志记录时，该缓冲区用来暂存一个Frame的长度的数据（Frame由日志数据和填充数据构成）。
	uint64buf []byte
}

func newEncoder(w io.Writer, prevCrc uint32, pageOffset int) *encoder {
	return &encoder{
		bw:  ioutil.NewPageWriter(w, walPageBytes, pageOffset),
		crc: crc.New(prevCrc, crcTable),
		// 1MB缓冲区
		buf:       make([]byte, 1024*1024),
		uint64buf: make([]byte, 8),
	}
}

// newFileEncoder使用当前文件偏移量为页面编写器创建一个新编码器。
func newFileEncoder(f *os.File, prevCrc uint32) (*encoder, error) {
	offset, err := f.Seek(0, io.SeekCurrent)
	if err != nil {
		return nil, err
	}
	return newEncoder(f, prevCrc, int(offset)), nil
}

// encoder.encode（）方法中，会对日志记录进行序列化，并完成8字节对齐
func (e *encoder) encode(rec *walpb.Record) error {
	e.mu.Lock()	// 加锁同步
	defer e.mu.Unlock()	// 整个方法结束后，释放锁
	// 计算crc校验码
	e.crc.Write(rec.Data)
	rec.Crc = e.crc.Sum32()
	var (
		data []byte
		err  error
		n    int
	)
	// 将待写入到日志记录进行序列化，如果日志记录太大，
	// 无法复用eneoder.buf这个缓冲区，则直接序列化
	if rec.Size() > len(e.buf) {
		data, err = rec.Marshal()
		if err != nil {
			return err
		}
	} else {	// 复用eneoder.buf这个缓冲区
		n, err = rec.MarshalTo(e.buf)
		if err != nil {
			return err
		}
		data = e.buf[:n]
	}
	// 计算序列化之后的数据长度，在eneodeFrarneSize（）方法中会完成8字节对齐，
	// 这里将真正的数据和填充数据看作一个Frame，返回值分别是整个Frame的长度，
	// 以及其中填充数据的长度
	lenField, padBytes := encodeFrameSize(len(data))
	// 将Frame的长度序列化到eneoder.uint64buf数组中，然后写入文件
	if err = writeUint64(e.bw, lenField, e.uint64buf); err != nil {
		return err
	}

	if padBytes != 0 {
		data = append(data, make([]byte, padBytes)...)	// 向data中写入填充字节
	}
	n, err = e.bw.Write(data)	// 将data中的序列化数据写入文件
	walWriteBytes.Add(float64(n))
	return err
}

func encodeFrameSize(dataBytes int) (lenField uint64, padBytes int) {
	lenField = uint64(dataBytes)
	// 强制8字节对齐，使长度永远不会被撕碎写入
	padBytes = (8 - (dataBytes % 8)) % 8
	if padBytes != 0 {
		lenField |= uint64(0x80|padBytes) << 56
	}
	return lenField, padBytes
}

func (e *encoder) flush() error {
	e.mu.Lock()
	n, err := e.bw.FlushN()
	e.mu.Unlock()
	walWriteBytes.Add(float64(n))
	return err
}

func writeUint64(w io.Writer, n uint64, buf []byte) error {
	// http:
	binary.LittleEndian.PutUint64(buf, n)
	nv, err := w.Write(buf)
	walWriteBytes.Add(float64(nv))
	return err
}
